package groupOne.app.DWS;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import groupOne.app.BaseAppOneStream;
import groupOne.bean.playVideoCount;
import groupOne.common.Constant;
import groupOne.util.AtguiguUtil;
import groupOne.util.DimAsyncFunction;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * @Author: LixuanFeng
 * @Date: 2022/9/2 19:55
 * @From: GoodGoodStudyDayDayUp
 *
 * 指标：
 *     视频播放次数    一条数据 +1
 *     累计播放时长    每条数据的play_sec 累加
 *     观看人数        人数要去重，状态维护上次观看日期，如果状态未null 或者 本条数据时间日期于状态不一致  人数+1
 *     人均观看时长     观看人数 / 播放时长
 *
 *      由于需求是要章节粒度，最细粒度为视频粒度，可以先求出视频粒度的聚合结果，
 *      后续再根据维度补充，把维度补充后 在doris 进行rollup获得更粗粒度的聚合结果
 *
 *      doris建表语句：
 *                  drop table if exists dws_trade_cart_add_uu_window;
 *  *              create table if not exists dws_play_times_people_count_window
 *  *              (
 *                         `stt`            DATETIME comment '窗口起始时间',
 *                        `edt`             DATETIME comment '窗口结束时间',
 *                        `cur_date`        DATE comment '当天日期',
 *                        `video_id`        VARCHAR(128) comment '视频id',
 *                        `chapter_id`      VARCHAR(128) comment  '章节id',
 *                        `chapter_name`    VARCHAR(128) comment  '章节name',
 *                        `uid`             VARCHAR(128) comment  '用户id',
 *                        `play_sec`        BIGINT replace comment '播放时长',
 *                        `play_ct`         BIGINT replace comment '播放次数',
 *                        `people_ct`       BIGINT replace comment '播放人数',
 *                        `playSecAvg`      BIGINT replace comment '人均播放时长'
 *                  )engine = olap aggregate key ( `stt`,`edt`,`cur_date`,`video_id`,`chapter_id`,`chapter_name`,`uid` )
 *                   comment "播放主题汇总表"
 *                   partition by range(`cur_date`)()
 *                   distributed by hash(`stt`) buckets 10 properties (
 *                  "replication_num" = "3",
 *                  "dynamic_partition.enable" = "true",
 *                  "dynamic_partition.time_unit" = "DAY",
 *                  "dynamic_partition.start" = "-1",
 *                  "dynamic_partition.end" = "1",
 *                  "dynamic_partition.prefix" = "par",
 *                  "dynamic_partition.buckets" = "10",
 *                  "dynamic_partition.hot_partition_num" = "1"
 *                );
 *
 */
public class DwsPlayTimesCountAvgChapterWindow_lixuanfeng extends BaseAppOneStream {

    public static void main(String[] args) {
        new DwsPlayTimesCountAvgChapterWindow_lixuanfeng().init(3121,2,
                "DwsPlayTimesCountAvgChapterWindow_lixuanfeng",
                Constant.TOPIC_DWD_TRAFFIC_PLAY_VIDEO);
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //1.转换格式并封装成pojo, 根据uid分组去重
        SingleOutputStreamOperator<playVideoCount> pojoStream = mapAndParseToPojo(stream);

        //2. 再根据 video_id 分组统计 ，开窗聚合
        SingleOutputStreamOperator<playVideoCount> aggStream = countByVideoWindow(pojoStream);

        //3.补充维度
        SingleOutputStreamOperator<playVideoCount> dimStream = addDim(aggStream);

        //4.写入doris
        writeToDoris(dimStream);

    }

    private void writeToDoris(SingleOutputStreamOperator<playVideoCount> dimStream) {

        Properties props = new Properties();
        props.setProperty("format", "json");
        props.setProperty("strip_outer_array", "true");
        String tableName= "edu.dws_play_times_people_count_window";


        dimStream.map(
                pojo -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;  // 转成json的时候, 属性名使用下划线
                    return JSON.toJSONString(pojo, config);
                })
                .addSink(DorisSink.sink(new DorisExecutionOptions.Builder()
                                .setBatchIntervalMs(2000L)
                                .setBatchSize(1024 * 1024)
                                .setEnableDelete(false)
                                .setMaxRetries(3)
                                .setStreamLoadProp(props)
                                .build(),
                        new DorisOptions.Builder()
                                .setFenodes(Constant.DORIS_HOST)
                                .setUsername("root")
                                .setPassword("aaaaaa")
                                .setTableIdentifier(tableName)
                                .build()));
    }


    private SingleOutputStreamOperator<playVideoCount> addDim(SingleOutputStreamOperator<playVideoCount> aggStream) {
        SingleOutputStreamOperator<playVideoCount> chapterId = AsyncDataStream.unorderedWait(
                aggStream,
                new DimAsyncFunction<playVideoCount>() {
                    @Override
                    public String getTable() {
                        return "DIM_VIDEO_INFO";
                    }

                    @Override
                    public String getId(playVideoCount input) {
                        return input.getVideoId();
                    }

                    @Override
                    public void addDim(JSONObject dim, playVideoCount input) {
                        input.setChapterId(dim.getString("CHAPTER_ID"));
                    }
                }
                , 60L,
                TimeUnit.SECONDS
        );

      return
        AsyncDataStream.unorderedWait(
                chapterId,
                new DimAsyncFunction<playVideoCount>() {
                    @Override
                    public String getTable() {
                        return "DIM_CHAPTER_INFO";
                    }

                    @Override
                    public String getId(playVideoCount input) {
                        return input.getChapterId();
                    }

                    @Override
                    public void addDim(JSONObject dim, playVideoCount input) {
                        input.setChapterId(dim.getString("CHAPTER_NAME"));
                    }
                }
                , 60L,
                TimeUnit.SECONDS
        );


    }


    private SingleOutputStreamOperator<playVideoCount> countByVideoWindow(SingleOutputStreamOperator<playVideoCount> pojoStream) {
        return
        pojoStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<playVideoCount>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                .withTimestampAssigner((pojo,ts)->pojo.getTs()))
                .keyBy(pojo->pojo.getVideoId())
                .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<playVideoCount>() {
                            @Override
                            public playVideoCount reduce(playVideoCount value1, playVideoCount value2) throws Exception {
                                value1.setPlaySec(value1.getPlaySec() + value2.getPlaySec());
                                value1.setPeopleCt(value1.getPeopleCt() + value2.getPeopleCt());

                                return value1;
                            }
                        },
                        new ProcessWindowFunction<playVideoCount, playVideoCount, String, TimeWindow>() {
                            @Override
                            public void process(String s, Context context, Iterable<playVideoCount> elements, Collector<playVideoCount> out) throws Exception {
                                playVideoCount pojo = elements.iterator().next();

                                String start = AtguiguUtil.toDate(context.window().getStart());
                                String end = AtguiguUtil.toDate(context.window().getEnd());
                                String curDate = AtguiguUtil.toDate(System.currentTimeMillis());

                                Long playSecAvg = pojo.getPlaySec() / pojo.getPeopleCt();

                                pojo.setStt(start);
                                pojo.setEdt(end);
                                pojo.setCurDate(curDate);
                                pojo.setPlaySecAvg(playSecAvg);

                                out.collect(pojo);
                            }
                        }
                );
                


    }

    private SingleOutputStreamOperator<playVideoCount> mapAndParseToPojo(DataStreamSource<String> stream) {
        return
             stream
                .map(jsonStr->{

                    JSONObject obj = JSON.parseObject(jsonStr);
                    String videoId = obj.getString("video_id");
                    Long playSec = obj.getLong("play_sec");
                    String uid = obj.getString("uid");
                    Long ts = obj.getLong("ts");
                    Long playCt=1L;

                    return   playVideoCount.builder()
                             .videoId(videoId)
                             .playSec(playSec)
                             .uid(uid)
                             .playCt(playCt)
                             .ts(ts)
                            .build();
                })
                .keyBy(pojo->pojo.getUid())
                .process(new KeyedProcessFunction<String, playVideoCount, playVideoCount>() {

                    private ValueState<String> lastPlayDate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastPlayDate = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastPlayDate", String.class));
                    }

                    @Override
                    public void processElement(playVideoCount value, Context ctx,
                                               Collector<playVideoCount> out) throws Exception {
                        String lastPD = lastPlayDate.value();
                        String thisPD = AtguiguUtil.toDate(value.getTs());

                        Long peopleCt=0L;
                        if ( lastPD==null || !thisPD.equals(lastPD)){
                            peopleCt=1L;
                            lastPlayDate.update(thisPD);
                        }

                        out.collect( playVideoCount.builder()
                                .peopleCt(peopleCt)
                                .build());
                    }
                });



    }
}
